package org.xq.softcup.mq.remoting.net.impl.netty.server;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xq.softcup.mq.remoting.net.params.XQRpcRequest;
import org.xq.softcup.mq.remoting.net.params.XQRpcResponse;
import org.xq.softcup.mq.remoting.provider.XQRpcProviderFactory;
import org.xq.softcup.mq.util.ThrowableUtil;

import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class NettyServerHandler extends SimpleChannelInboundHandler<XQRpcRequest> {
    private XQRpcProviderFactory xqRpcProviderFactory;
    private ThreadPoolExecutor serverHandlerPool;

    public NettyServerHandler(XQRpcProviderFactory xqRpcProviderFactory, ThreadPoolExecutor serverHandlerPool) {
        this.xqRpcProviderFactory = xqRpcProviderFactory;
        this.serverHandlerPool = serverHandlerPool;
    }

    public void channelRead0(final ChannelHandlerContext ctx, final XQRpcRequest xqRpcRequest) throws Exception {
        try {
            this.serverHandlerPool.execute(new Runnable() {
                public void run() {
                    XQRpcResponse xqRpcResponse = NettyServerHandler.this.xqRpcProviderFactory.invokeService(xqRpcRequest);
                    ctx.writeAndFlush(xqRpcResponse);
                }
            });
        } catch (Exception e) {
            XQRpcResponse xqRpcResponse = new XQRpcResponse();
            xqRpcResponse.setRequestId(xqRpcRequest.getRequestId());
            xqRpcResponse.setErrorMsg(ThrowableUtil.toString(e));

            ctx.writeAndFlush(xqRpcResponse);
        }
    }

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("Netty服务端连接异常", cause);
        ctx.close();
    }
}
